package com.surfilter.massdata.spark.output.ipcheck;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import com.act.sparkanalyz.output.impl.HiveOutput;
import com.surfilter.massdata.spark.util.JdbcUtil;

public class IpCheckOutTest extends HiveOutput {

	private static final long serialVersionUID = 1L;
	private static final Log log = LogFactory.getLog(IpCheckOutTest.class);
	private String columns;
	private String tableName;
	private String values;
	@Override
	public void write(DataFrame result, Map<String, String> commandMap) {
		log.info("===========IpCheckOutTest begin============");
//		List<Row> list = result.collectAsList();
//		try {
//			insertTable(list);
//		} catch (SQLException e) {
//			// TODO Auto-generated catch block
//			e.printStackTrace();
//		}
		log.info("===========IpCheckOutTest end============");
	}
	private void insertTable(List<Row> list) throws SQLException {
		String[] columnsSplit = columns.split(",");
		Connection conn = null;
		PreparedStatement ps = null;
		String sql = "insert into "+tableName+ "("+columns+") "+values;
		 try {
			Class.forName("oracle.jdbc.driver.OracleDriver");
		} catch (ClassNotFoundException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}// 加载Oracle驱动程序
	    String url = "jdbc:oracle:" + "thin:@127.0.0.1:1521:orcl";// 127.0.0.1是本机地址，XE是精简版Oracle的默认数据库名
        String user = "orcl";// 用户名,系统默认的账户名
        String password = "orcl";// 你安装时选设置的密码
        conn = DriverManager.getConnection(url, user, password);// 获取连接
//		conn = JdbcUtil.getConnection();
		try {
			ps = conn.prepareStatement(sql);
			conn.setAutoCommit(false);
			int n = 0;
			for (Row row : list) {
				for (int i = 0; i < row.size(); i++) {
					System.err.println(columnsSplit[i]+"\t"+row.getAs(columnsSplit[i]).toString());
					ps.setObject(i + 1, row.getAs(columnsSplit[i]));
				}
				ps.addBatch();

				if (++n % 5000 == 0) {
					ps.executeBatch();
					conn.commit();
				}
			}
			ps.executeBatch();
			conn.commit();
		} catch (Exception e) {
			log.error("========="+tableName+"========");
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}finally{
			JdbcUtil.close(ps,conn);
		}
	}
}
